package com.example.hadoop.mapreduce.kmeans;

import com.example.hadoop.others.Helper;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.net.URISyntaxException;

public class KmeansMapper extends Mapper<LongWritable, Text, IntWritable, Text> {

    private int nClusters;
    private double[][] centers;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        FileSystem fileSystem = null;
        try {
            fileSystem = Helper.getFileSystem();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
        String jobName = context.getJobName();
        centers = Helper.getCenters(fileSystem);
        nClusters = centers.length;
        super.setup(context);
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // super.map(key, value, context);
        double[] data = new double[2];
        double minDistance = Double.MAX_VALUE;
        double distance = 0.0;
        int index = 999;
        String[] content = value.toString().trim().split(",");
        data[0] = Double.valueOf(content[1]);
        data[1] = Double.valueOf(content[2]);
        for (int i = 0; i < nClusters; i++) {
            distance = Helper.calDistance(data, centers[i]);
            if (distance <= minDistance) {
                minDistance = distance;
                index = i;
            }
        }
        context.write(new IntWritable(index), value);
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        super.cleanup(context);
    }

    @Override
    public void run(Context context) throws IOException, InterruptedException {
        super.run(context);
    }
}
